//
// Copyright (C) 2024 Kumo group inc.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//

#include <krpc/rpc/restful_client.h>
#include <turbo/strings/match.h>
#include <turbo/flags/flag.h>

TURBO_FLAG(int32_t, restful_connect_timeout_ms, 50, "RPC timeout in milliseconds");
TURBO_FLAG(int32_t, restful_timeout_ms, 500, "RPC timeout in milliseconds");
TURBO_FLAG(int32_t, restful_max_retry, 3, "Max retries(not including the first RPC)");
TURBO_FLAG(int32_t, restful_concurrency, 1000, "Max number of http calls in parallel");

namespace krpc {

    int get_schema_default_port(const std::string &schema) {
        if (turbo::equals_ignore_case(schema, "http")) {
            return 80;
        } else if (turbo::equals_ignore_case(schema, "https")) {
            return 443;
        } else if (turbo::equals_ignore_case(schema, "redis")) {
            return 6379;
        }
        return -1;
    }

    turbo::Result<kutil::EndPoint> parse_endpoint(const turbo::Uri &uri) {
        auto host = uri.host();
        int port = uri.port();
        if (host.empty()) {
            return turbo::invalid_argument_error("No Host");
        }
        if (port == -1) {
            auto schema = uri.scheme();
            if (schema.empty()) {
                return turbo::invalid_argument_error("No schema");
            }
            port = get_schema_default_port(schema);
            if (port == -1) {
                return turbo::invalid_argument_error("unknown schema and not give port");
            }
        }
        kutil::EndPoint ret;
        if (kutil::str2endpoint(host.c_str(), port, &ret) < 0 &&
            kutil::hostname2endpoint(host.c_str(), port, &ret) < 0) {
            return turbo::invalid_argument_error("convert to endpoint fail host: %s port: %d", host, port);
        }
        return ret;
    }

    turbo::Result<kutil::EndPoint> parse_endpoint(const URI &uri) {
        if (!uri.status().ok()) {
            return turbo::invalid_argument_error(uri.status().error_cstr());
        }
        auto host = uri.host();
        int port = uri.port();
        if (host.empty()) {
            return turbo::invalid_argument_error("No Host");
        }
        if (port == -1) {
            auto schema = uri.scheme();
            if (schema.empty()) {
                return turbo::invalid_argument_error("No schema");
            }
            port = get_schema_default_port(schema);
            if (port == -1) {
                return turbo::invalid_argument_error("unknown schema and not give port");
            }
        }
        kutil::EndPoint ret;
        if (kutil::str2endpoint(host.c_str(), port, &ret) < 0 &&
            kutil::hostname2endpoint(host.c_str(), port, &ret) < 0) {
            return turbo::invalid_argument_error("convert to endpoint fail host: %s port: %d", host, port);
        }
        return ret;
    }

    turbo::Result<kutil::EndPoint> parse_endpoint(const std::string &uri) {
        turbo::Uri parsed;
        std::string err;
        if (!parsed.parse(uri, &err)) {
            return turbo::invalid_argument_error(err);
        }
        return parse_endpoint(parsed);
    }


    class OnRestfulCallback : public google::protobuf::Closure {
    public:
        void Run();

    public:
        std::shared_ptr<RestfulTask> task;
        std::function<void(const std::shared_ptr<RestfulTask>)> callback;
        RestfulClient *client;
    };

    void OnRestfulCallback::Run() {
        std::unique_ptr<OnRestfulCallback> delete_self(this);
        callback(task);
        client->decrement();
    }

    class OnFutureCallback : public google::protobuf::Closure {
    public:
        void Run();

    public:
        std::shared_ptr<RestfulTask> task;
        RestfulClient *client;
    };

    void OnFutureCallback::Run() {
        std::unique_ptr<OnFutureCallback> delete_self(this);
        client->decrement();
        LOG(INFO) << "OnFutureCallback::Run";
        task->signal();
    }

    ChannelOptions RestfulClient::get_default_options() {
        krpc::ChannelOptions options;
        options.protocol = krpc::PROTOCOL_HTTP;
        options.connect_timeout_ms = turbo::get_flag(FLAGS_restful_connect_timeout_ms);
        options.timeout_ms = turbo::get_flag(FLAGS_restful_timeout_ms);
        options.max_retry = turbo::get_flag(FLAGS_restful_max_retry);
        return options;
    }

    RestfulTask RestfulClient::get(const std::string &url) {
        RestfulTask task;
        task.set_url(url);
        get(&task);
        return task;
    }

    void RestfulClient::get(RestfulTask *task) {
        auto &uri = task->url();
        if (uri.empty()) {
            task->set_status(turbo::invalid_argument_error("your must set url for the request"));
            return;
        }
        auto re = parse_endpoint(uri);
        if (!re.ok()) {
            task->set_status(re.status());
            return;
        }
        while (increment()
               > _concurrent_limit) {
            decrement();
            kthread_usleep(5000);
        }
        krpc::Channel channel;
        auto options = get_default_options();
        if (channel.Init(re.value(), &options) != 0) {
            auto str = turbo::str_format("Fail to create channel to url = %s", uri);
            LOG(ERROR) << str;
            task->set_status(turbo::invalid_argument_error(str));
            return;
        }
        channel.CallMethod(nullptr, task->controller(), nullptr, nullptr, nullptr);
    }

    void RestfulClient::async_get(const std::string &url,
                                  std::function<void(const std::shared_ptr<RestfulTask> &task)> callback) {
        auto task = std::make_shared<RestfulTask>();
        task->set_url(url);
        async_get(task, std::forward<std::function<void(const std::shared_ptr<RestfulTask> &)>>(callback));
    }

    void RestfulClient::async_get(const std::shared_ptr<RestfulTask> &task,
                                  std::function<void(const std::shared_ptr<RestfulTask> &task)> callback) {
        DCHECK(!callback) << "call back empty";
        auto &uri = task->url();
        if (uri.empty()) {
            task->set_status(turbo::invalid_argument_error("your must set url for the request"));
            callback(task);
            return;
        }
        auto re = parse_endpoint(uri);
        if (!re.ok()) {
            task->set_status(re.status());
            return;
        }
        while (increment()
               > _concurrent_limit) {
            decrement();
            kthread_usleep(5000);
        }
        krpc::Channel channel;
        auto options = get_default_options();
        if (channel.Init(re.value(), &options) != 0) {
            auto str = turbo::str_format("Fail to create channel to url = %s", uri);
            LOG(ERROR) << str;
            task->set_status(turbo::invalid_argument_error(str));
            callback(task);
            return;
        }

        auto done = new OnRestfulCallback;
        done->callback = std::move(callback);
        done->task = task;
        channel.CallMethod(nullptr, task->controller(), nullptr, nullptr, done);

    }

    void RestfulClient::async_get(const std::shared_ptr<RestfulTask> &task) {
        auto &uri = task->url();
        if (uri.empty()) {
            task->set_status(turbo::invalid_argument_error("your must set url for the request"));
            return;
        }
        auto re = parse_endpoint(uri);
        if (!re.ok()) {
            task->set_status(re.status());
            return;
        }
        while (increment()
               > _concurrent_limit) {
            LOG(WARNING) << "to many: " << decrement();
            kthread_usleep(5000);
        }
        task->make_event();
        krpc::Channel channel;
        auto options = get_default_options();
        if (channel.Init(re.value(), &options) != 0) {
            auto str = turbo::str_format("Fail to create channel to url = %s", uri);
            LOG(ERROR) << str;
            task->set_status(turbo::invalid_argument_error(str));
            return;
        }

        auto done = new OnFutureCallback;
        done->task = task;
        done->client = this;
        channel.CallMethod(nullptr, task->controller(), nullptr, nullptr, done);
        return;
    }

    std::shared_ptr<RestfulTask> RestfulClient::async_get(const std::string &url) {
        auto task = std::make_shared<RestfulTask>();
        task->set_url(url);
        async_get(task);
        return task;
    }

}  // namespace krpc
